package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.util.ZkUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
public class RpcClientProxy {

    private static final List<RpcClient> clients = new ArrayList<>();
    private Map<String, String[]> connectMap = new HashMap<>();
    private static ZkUtil zkUtil=new ZkUtil();


    public static Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        //2.创建RpcClient对象
                        String nodeAddress = loadBalance();
                        String ipAndPort = nodeAddress.split("/", -1)[2];
                        String ip=ipAndPort.split(":", -1)[0];
                        int port=Integer.parseInt(ipAndPort.split(":", -1)[1]);
                         RpcClient rpcClient = new RpcClient(ip, port);
                        try {
                            //请求时间
                            long requestTime = System.currentTimeMillis();
                            //3.发送消息
                            String jsonString = JSON.toJSONString(rpcRequest);
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));

                            //获取响应时间
                            long responseTime = System.currentTimeMillis();
                            long time = (responseTime - requestTime)/1000;
                            zkUtil.updateNodeData(nodeAddress, (responseTime+"-"+time));
                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            //4.返回结果
                            Object result = rpcResponse.getResult();
                            return JSON.parseObject(result.toString(), method.getReturnType());
                        } catch (Exception e) {
                            throw e;
                        } finally {
                            rpcClient.close();
                        }

                    }
                });
    }

    // 监听服务
    public void watch() {

        try {
            TreeCache treeCache =
                    new TreeCache(ZkUtil.getClient(), "/lg-rpc");
            TreeCacheListener treeCacheListener =
                    new TreeCacheListener() {
                        @Override
                        public void childEvent(CuratorFramework client,
                                               TreeCacheEvent event) throws Exception {
                            ChildData data = event.getData();
                            if (data == null || data.getPath().equals("/lg-rpc")) {
                                return;
                            }

                            String nodeData = new String(data.getData());

                            switch (event.getType()) {
                                case NODE_ADDED:
                                    System.out.println("节点增加，path：" + data.getPath() + "，节点内容：" + nodeData);

                                    String[] ipAndPort = nodeData.split(":", -1);
                                    connectMap.put(data.getPath(), ipAndPort);
                                    break;
                                case NODE_REMOVED:
                                    System.out.println("节点删除，path：" + data.getPath() + "，节点内容：" + nodeData);

                                    connectMap.remove(data.getPath());

                                    break;
                                default:
                                    break;
                            }

                        }
                    };
            treeCache.getListenable().addListener(treeCacheListener);
            treeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static String loadBalance() {
        //192.128.174.128:8088-2
        float minTime = 5f;
        String minNodeAddress = "";
        ZkUtil zkUtil = new ZkUtil();
        List<String> arrayList = new ArrayList<>();
        Map<String, String> addressList = zkUtil.getAddressList();
        Set<Map.Entry<String, String>> entrySet = addressList.entrySet();
        for (Map.Entry<String, String> entry : entrySet) {
            String nodeAddress = entry.getKey();
            String nodeValue = entry.getValue();
            float serverTime = Float.parseFloat(nodeValue.split("-",-1)[1]);
            //如果小于最小时间则更新最小时间，并且将之前的list清空，重新添加
            if (serverTime < minTime) {
                minTime = serverTime;
                arrayList = new ArrayList<>();
                arrayList.add(entry.getKey());
            }
            //如果等于最小时间，则将其服务添加到list中，方便后续随机调用
            if (serverTime == minTime) {
                arrayList.add(entry.getKey());
            }
        }
        if (arrayList.size() > 1) {
            minNodeAddress = arrayList.get(new Random().nextInt(addressList.size()));
            System.out.println("节点" + minNodeAddress + "的响应时间最短，为" + minTime);
        }
        if (arrayList.size() == 1) {
            minNodeAddress = arrayList.get(0);
        }
        //不存在arrayList.size()==0的情况，因为超过5s的会删除掉
        return minNodeAddress;
    }
}
